package org.example.rabbitmq;

import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;

import java.nio.charset.StandardCharsets;

public class MqttSubscriber {

    public static void main(String[] args) {
        // RabbitMQ MQTT 地址和端口
        String host = "192.168.0.227";
        int port = 1883;

        // MQTT 主题（需与发送方一致）
        String topicFilter = "test/topic";

        // 创建客户端 ID（可随机）
        String clientId = "JavaMqttSubscriber";

        // 构建异步客户端
        Mqtt3AsyncClient client = Mqtt3Client.builder()
                .identifier(clientId)
                .serverHost(host)
                .serverPort(port)
                .buildAsync();

        // 设置连接成功后的回调
        client.connectWith()
                .simpleAuth()
                .username("mq")
                .password("123456".getBytes())
                .applySimpleAuth()
                .send()
                .thenAccept(connAck -> {
                    System.out.println("MQTT client connected successfully.");

                    // 订阅主题
                    client.subscribeWith()
                            .topicFilter(topicFilter)
                            .qos(MqttQos.AT_LEAST_ONCE)
                            .callback(publish -> {
                                String receivedMessage = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
                                System.out.println("Received message on topic [" + publish.getTopic() + "]: " + receivedMessage);
                            })
                            .send();
                })
                .exceptionally(throwable -> {
                    System.err.println("Connection failed: " + throwable);
                    return null;
                });

        // 阻塞主线程以保持订阅监听
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 断开连接并关闭客户端
        client.disconnect();
    }
}
